package yeliuhuishi.TopMovie;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import yeliuhuishi.File.DelAllFile;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;

/**
 * 统计每个用户的前5部电影
 * order desc
 */
public class RateMapperReducer {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        String input = "src/main/java/yeliuhuishi/TopMovie/input";
        String output = "src/main/java/yeliuhuishi/TopMovie/output";
        DelAllFile.delAllFile(new File(output));
        Configuration conf = new Configuration();

        conf.set("topN", "5");

        Job job = Job.getInstance(conf, "rate");

        job.setMapperClass(RateMapper.class);
        job.setReducerClass(RateReducer.class);
        job.setJarByClass(RateMapperReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(RateBean.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(RateBean.class);

        FileInputFormat.setInputPaths(job, new Path(input));//input
        //设置输出路径
        FileOutputFormat.setOutputPath(job, new Path(output));//output
        //  提交
        //  没有返回值,不知道是否成功,所以不用
        //  job.submit();

        System.out.println(job.waitForCompletion(true) ? "成功" : "失败");


    }

    public static class RateMapper extends Mapper<LongWritable, Text, Text, RateBean> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            RateBean rateBean = new RateBean();
            String[] s = value.toString().split(" ");
            rateBean.setMovie(s[0]);
            rateBean.setRate(s[1]);
            rateBean.setTimeStamp(s[2]);
            rateBean.setUid(s[3]);
            context.write(new Text(s[3]), rateBean);
        }
    }

    public static class RateReducer extends Reducer<Text, RateBean, Text, RateBean> {
        @Override
        protected void reduce(Text key, Iterable<RateBean> values, Context context) throws IOException, InterruptedException {
            // 定义了一个list，对list排序
            ArrayList<RateBean> rateBeans = new ArrayList<RateBean>();
            Configuration conf = context.getConfiguration();

            // 默认设置为5
            int topN = conf.getInt("topN", 5);
            for (RateBean value : values) {
                RateBean rateBean = new RateBean();
                rateBean.setMovie(value.getMovie());
                rateBean.setRate(value.getMovie());
                rateBean.setTimeStamp(value.getTimeStamp());
                rateBean.setUid(value.getUid());
                rateBeans.add(rateBean);
            }
            Collections.sort(rateBeans);
            for (int i = 0; i < topN; i++) {
                context.write(key, rateBeans.get(i));
            }
        }
    }
}
